# -*- coding: utf-8 -*-
# This code written by ChanGOGOGO
"""
对通话记录进行大模型质检
"""
import pymysql
from uuid import uuid1
from concurrent.futures import ThreadPoolExecutor, as_completed
from util import (
    read_config,
    logging,
    data2beautify,
    are_contain_str,
    query_es,
    time,
    post_json,
    write_excel2,
    get_time_desc,
    db_ip,
    db_port,
    db_username,
    db_password,
    es_ip,
    es_port,
)


class ExportLLMInspection:
    def __init__(self, ui_data):
        self.large_model_config = read_config("./config/large_model.yaml")
        self.large_model_address = self.large_model_config.get("large_model_address")
        self.max_thread = self.large_model_config.get(
            "max_thread"
        )  # 调用大模型最大线程数
        self.exclude_show_type = self.large_model_config.get(
            "exclude_show_type"
        )  # 将带有指定标识的对话轮次信息剔除，不参与大模型质检
        self.exclude_company_ids = self.large_model_config.get(
            "exclude_company_ids"
        )  # 不需要参与质检的账号
        self.exclude_output_parameters = self.large_model_config.get(
            "exclude_output_parameters"
        )  # 不需要参与质检的通话记录出参关键字
        self.data_date = ui_data["data_date"]
        self.business = ui_data.get("business")
        self.data_type = ui_data.get("data_type")
        self.call_type = ui_data.get("call_type")
        self.start_time = ui_data.get("start_time")
        self.end_time = ui_data.get("end_time")
        self.company_id = ui_data.get("company_id")
        self.db_name = f"yicall_{self.company_id}"
        self.table_index = ui_data.get("business_id")
        self.conn = pymysql.connect(
            host=db_ip,
            port=db_port,
            user=db_username,
            passwd=db_password,
            charset="utf8",
        )
        self.cursor = self.conn.cursor()
        self.scene_info = self.query_scene()
        self.ivc_engine_info = self.query_ivc_engine()
        self.callout_task_engine_name_info = self.query_callout_task_engine_name_info()
        self.callin_task_engine_name_info = self.query_callin_task_engine_name_info()

    def query_business(self):
        """查询所有的业务信息：company_id、business_id、business_name"""
        if self.exclude_company_ids:  # 需要剔除一些不参与统计的company_id
            exclude_company_ids_len = len(self.exclude_company_ids)
            if exclude_company_ids_len == 1:
                exclude_company_condition = (
                    f"and a.company_id != {self.exclude_company_ids[0]}"
                )
            else:
                new_exclude_company_ids = list(map(str, self.exclude_company_ids))
                exclude_company_condition = (
                    f"and a.company_id not in ({','.join(new_exclude_company_ids)})"
                )
        else:
            exclude_company_condition = ""
        sql = f"""select * from (select a.company_id,b.bid from yicall_business.yicall_aiforce_user a RIGHT JOIN
        yicall_business.yicall_aiforce_instance b on a.company_id=b.company_id where a.company_id>=1000
        {exclude_company_condition} ORDER BY a.company_id) temp GROUP BY company_id,bid;"""
        # logging.info(f'查询所有的业务信息SQL：\n{sql}')
        self.cursor.execute(sql)
        company_business_id = self.cursor.fetchall()
        company_business_info = list()
        for x in company_business_id:
            company_id, business_id = x
            sql = f"""select pid,business_id,business_name from yicall_{company_id}.call_business
            where pid={company_id} and business_id={business_id};"""
            self.cursor.execute(sql)
            result = self.cursor.fetchall()
            company_business_info.append(result[0])
        logging.info(
            f"获取到所有业务company_id、business_id、business_name：\n{data2beautify(company_business_info)}"
        )
        return company_business_info

    def query_scene(self):
        """查询场景信息"""
        sql = f"""select scene_id,scene_name from {self.db_name}.b_{self.table_index}_call_scene
        order by scene_id desc;"""
        self.cursor.execute(sql)
        result = self.cursor.fetchall()
        scene_info = {}
        if len(result) > 0:
            scene_info = {key: value for key, value in result}
        return scene_info

    def query_ivc_engine(self):
        """查询ivc引擎信息"""
        sql = f"""select engine_id, engine_name from yicall_business.ivc_engine_config 
        where company_id={self.company_id} and business_id={self.table_index};"""
        logging.info(f"查询ivc引擎sql：\n{sql}")
        self.cursor.execute(sql)
        result = self.cursor.fetchall()
        ivc_engine_info = {}
        if len(result) > 0:
            ivc_engine_info = {key: value for key, value in result}
        logging.info(f"查询到ivc引擎信息：\n{data2beautify(ivc_engine_info)}")
        return ivc_engine_info

    def query_callout_task_engine_name_info(self):
        """查询呼出任务对应的引擎名字"""
        sql = f"""select task_id,JSON_EXTRACT(asr_properties, '$.engineId') as engine_id 
        from {self.db_name}.b_{self.table_index}_callout_task;"""
        logging.info(f"查询呼出任务id和asr引擎信息sql：\n{sql}")
        self.cursor.execute(sql)
        result = self.cursor.fetchall()
        task_engine_id_info = {}
        callout_task_engine_name_info = {}
        if len(result) > 0:
            task_engine_id_info = {key: value for key, value in result}
        logging.info(f"任务与引擎ID信息：\n{data2beautify(task_engine_id_info)}")
        if task_engine_id_info:
            for task_id, engine_id in task_engine_id_info.items():
                if engine_id is None:
                    engine_name = "未知"
                else:
                    engine_name = self.ivc_engine_info.get(int(engine_id))
                callout_task_engine_name_info[task_id] = engine_name
        logging.info(
            f"查询到呼出任务使用到的引擎名称：\n{data2beautify(callout_task_engine_name_info)}"
        )
        return callout_task_engine_name_info

    def query_callin_task_engine_name_info(self):
        """查询呼入任务对应的引擎名字"""
        sql = f"""select task_id,JSON_EXTRACT(asr_properties, '$.engineId') as engine_id 
        from {self.db_name}.b_{self.table_index}_callout_task_in;"""
        logging.info(f"查询呼出任务id和asr引擎信息sql：\n{sql}")
        self.cursor.execute(sql)
        result = self.cursor.fetchall()
        task_engine_id_info = {}
        callin_task_engine_name_info = {}
        if len(result) > 0:
            task_engine_id_info = {key: value for key, value in result}
        if task_engine_id_info:
            for task_id, engine_id in task_engine_id_info.items():
                if engine_id is None:
                    engine_name = "未知"
                else:
                    engine_name = self.ivc_engine_info.get(int(engine_id))
                callin_task_engine_name_info[task_id] = engine_name
        logging.info(
            f"查询到呼入任务使用到的引擎名称：\n{data2beautify(callin_task_engine_name_info)}"
        )
        return callin_task_engine_name_info

    def close_db(self):
        """关闭数据库连接"""
        self.cursor.close()
        self.conn.close()

    def query_callout_info(self):
        """查询呼出通话记录信息"""
        es_index = (
            f"{self.db_name}_{self.table_index}_callout_session_ach_{self.data_date}"
        )
        body = {
            "query": {
                "bool": {
                    "must": [
                        {"term": {"calloutConnectStatus": 100}},
                        {"range": {"dialogCustomerTurnCount": {"gte": 1}}},
                        {
                            "range": {
                                "calloutFinishTime": {
                                    "gte": self.start_time,
                                    "lte": self.end_time,
                                }
                            }
                        },
                    ]
                }
            },
            "sort": [{"calloutDialTime": {"order": "asc"}}],
        }
        fields = [
            "customerCaseId",
            "dialNumber",
            "calloutDialTime",
            "sceneId",
            "dialogUserPrefer",
            "calloutLastNodeName",
            "dialogCustomerTurnCount",
            "dialogBotTurnCount",
            "dialogDuration",
            "dialogSignoffTrigger",
            "turnList",
            "taskId",
            "sessionAnalyseInfo",
        ]
        es_data = query_es(es_index, body, fields, es_ip, es_port)
        finally_es_data = list()
        # 通话场景分析出参中包含某些关键字的话，通话无需参与大模型质检
        for single_es_data in es_data:
            session_analyse_info = single_es_data.get("sessionAnalyseInfo")
            output_parameter_value = session_analyse_info.values()
            find_exclude_output_parameter = are_contain_str(
                output_parameter_value, self.exclude_output_parameters
            )
            if find_exclude_output_parameter is False:
                finally_es_data.append(single_es_data)
        return finally_es_data

    def query_callin_info(self):
        """查询呼入通话记录信息"""
        es_index = (
            f"{self.db_name}_{self.table_index}_callout_session_in_ach_{self.data_date}"
        )
        body = {
            "query": {
                "bool": {
                    "must": [
                        {"term": {"calloutConnectStatus": 100}},
                        {"range": {"dialogCustomerTurnCount": {"gte": 1}}},
                        {
                            "range": {
                                "calloutFinishTime": {
                                    "gte": self.start_time,
                                    "lte": self.end_time,
                                }
                            }
                        },
                    ]
                }
            },
            "sort": [{"calloutDialTime": {"order": "asc"}}],
        }
        fields = [
            "customerCaseId",
            "dialNumber",
            "calloutDialTime",
            "sceneId",
            "dialogUserPrefer",
            "calloutLastNodeName",
            "dialogCustomerTurnCount",
            "dialogBotTurnCount",
            "dialogDuration",
            "dialogSignoffTrigger",
            "turnList",
            "taskId",
            "sessionAnalyseInfo",
        ]
        es_data = query_es(es_index, body, fields, es_ip, es_port)
        finally_es_data = list()
        # 通话场景分析出参中包含某些关键字的话，通话无需参与大模型质检
        for single_es_data in es_data:
            session_analyse_info = single_es_data.get("sessionAnalyseInfo")
            output_parameter_value = session_analyse_info.values()
            find_exclude_output_parameter = are_contain_str(
                output_parameter_value, self.exclude_output_parameters
            )
            if find_exclude_output_parameter is False:
                finally_es_data.append(single_es_data)
        return finally_es_data

    def quality_inspection(self, dialog_turn_info: list, dialog_time):
        """调用大模型接口进行质检"""
        data = {"id": str(uuid1()), "dialog": dialog_turn_info, "time": dialog_time}
        url = f"{self.large_model_address}/chat/v1/see/aibpo"
        try:
            res = post_json(url, data)
            return res
        except Exception as reason:
            logging.error(f"请求大模型质检接口异常：{reason}")
            return {}

    def handle_single_call_data(self, call_data: dict, call_type="callout"):
        """单条通话记录进行质检"""
        case_id = call_data.get("customerCaseId")
        user_phone = call_data.get("dialNumber")
        dial_time = call_data.get("calloutDialTime")
        scene_id = call_data.get("sceneId")
        scene_name = self.scene_info.get(int(scene_id))
        user_prefer = call_data.get("dialogUserPrefer")
        hang_up_node = call_data.get("calloutLastNodeName")
        user_turn_count = call_data.get("dialogCustomerTurnCount")
        bot_turn_count = call_data.get("dialogBotTurnCount")
        dialog_duration = int(call_data.get("dialogDuration")) / 1000
        dialog_sign_off_trigger = call_data.get("dialogSignoffTrigger")
        if dialog_sign_off_trigger == "1":
            hang_up_person = "客户"
        elif dialog_sign_off_trigger == "2":
            hang_up_person = "机器人"
        else:
            hang_up_person = "未接通"
        turn_list = call_data.get("turnList")
        task_id = call_data.get("taskId")
        if call_type == "callout":
            asr_name = self.callout_task_engine_name_info.get(int(task_id))
        else:
            asr_name = self.callin_task_engine_name_info.get(int(task_id))
        unsorted_turn_list = list()
        for turn in turn_list:
            show_type = turn.get("showType")
            if show_type not in self.exclude_show_type:
                start_timestamp = int(turn.get("startTimestamp"))
                voice_text = turn.get("voiceText")
                turn_holder = turn.get("turnHolder")
                unsorted_turn_list.append((start_timestamp, turn_holder, voice_text))
        sorted_turn_list = sorted(unsorted_turn_list, key=lambda x: x[0])
        excel_dialogs = list()
        large_model_dialogs = list()
        for y in sorted_turn_list:
            start_timestamp, turn_holder, voice_text = y
            if turn_holder == "1":
                role = "assistant"
                dm_role = "机器人"
            else:
                role = "user"
                dm_role = "客  户"
            large_model_dialogs.append({"role": role, "content": voice_text})
            excel_dialogs.append(f"【{dm_role}】:{voice_text}")
        if excel_dialogs and large_model_dialogs:
            excel_dialogs_content = "\n".join(excel_dialogs)
            quality_inspection_result = self.quality_inspection(
                large_model_dialogs, dial_time
            )
            code = quality_inspection_result.get("code")
            if code == 0:
                result = quality_inspection_result.get("data").get("result")
                understand_and_respond_ability = ""
                understand_and_respond_reason = ""
                emotion_recognition_ability = ""
                emotion_recognition_reason = ""
                appeasement_ability = ""
                appeasement_reason = ""
                asr_recognition_ability = ""
                asr_recognition_reason = ""
                for data in result:
                    name = data.get("name")
                    if name == "理解与回复能力":
                        understand_and_respond_ability = data.get("result")
                        understand_and_respond_reason = data.get("reason")
                    elif name == "情绪识别能力":
                        emotion_recognition_ability = data.get("result")
                        emotion_recognition_reason = data.get("reason")
                    elif name == "安抚能力":
                        appeasement_ability = data.get("result")
                        appeasement_reason = data.get("reason")
                    elif name == "ASR识别-转译错误":
                        asr_recognition_ability = data.get("result")
                        asr_recognition_reason = data.get("reason")
                row_data = [
                    case_id,
                    task_id,
                    asr_name,
                    user_phone,
                    dial_time,
                    scene_name,
                    user_prefer,
                    hang_up_node,
                    user_turn_count,
                    bot_turn_count,
                    dialog_duration,
                    hang_up_person,
                    excel_dialogs_content,
                    understand_and_respond_ability,
                    understand_and_respond_reason,
                    emotion_recognition_ability,
                    emotion_recognition_reason,
                    appeasement_ability,
                    appeasement_reason,
                    asr_recognition_ability,
                    asr_recognition_reason,
                ]
                return row_data
        else:
            return

    def export(self):
        start_time = time.time()
        all_call_datas = list()  # 所有拨打记录
        callout_datas = self.query_callout_info()  # 先查询呼出数据
        all_call_datas.extend(callout_datas)
        callin_datas = self.query_callin_info()  # 再查询呼入数据
        all_call_datas.extend(callin_datas)  # 呼入和呼出合并
        end_time = time.time()
        take_time = end_time - start_time
        logging.info(
            f"查询{self.business}在{self.start_time}至{self.end_time}之间的拨打数据共计耗时{take_time:.2f}秒。"
        )
        call_call_datas_count = len(all_call_datas)
        if call_call_datas_count == 0:
            msg = (
                f"<font color=green>无法导出“{self.business}”的【{self.data_type}】，"
                f"原因：\n在{self.start_time}至{self.end_time}没有通话数据！</font>"
            )
            logging.info(msg)
            return {"code": 1, "excel_names": None, "msg": msg}
        else:
            logging.info(
                f"{self.db_name}下的{self.table_index}在{self.start_time}至{self.end_time}之间"
                f"存在{call_call_datas_count}条拨打数据。"
            )
            excel_data = list()
            start_time = time.time()
            with ThreadPoolExecutor(max_workers=self.max_thread) as executor:
                # 提交任务到线程池
                futures = [
                    executor.submit(self.handle_single_call_data, data, "callin")
                    for data in callin_datas
                ]
                # 处理完成的任务
                for future in as_completed(futures):
                    try:
                        row_data = future.result()  # 获取线程任务结果
                        if row_data:
                            excel_data.append(row_data)
                    except Exception as e:
                        logging.error(f"Exception occurred: {e}")
            with ThreadPoolExecutor(max_workers=self.max_thread) as executor:
                # 提交任务到线程池
                futures = [
                    executor.submit(self.handle_single_call_data, data, "callout")
                    for data in callout_datas
                ]
                # 处理完成的任务
                for future in as_completed(futures):
                    try:
                        row_data = future.result()  # 获取线程任务结果
                        if row_data:
                            excel_data.append(row_data)
                    except Exception as e:
                        logging.error(f"Exception occurred: {e}")
            end_time = time.time()
            take_time = int(end_time - start_time)
            take_time_desc = get_time_desc(take_time)
            logging.info(
                f'对"{self.business}"在{self.start_time}至{self.end_time}之间的{call_call_datas_count}条拨打数据质检'
                f"共计耗时{take_time_desc}。"
            )
            sheet_name = f"{self.business}_14B"
            excel_title = [
                "caseID",
                "任务ID",
                "ASR资源",
                "客户手机",
                "拨打时间",
                "所属场景",
                "客户意向",
                "挂断节点",
                "客户对话轮次",
                "机器人对话轮次",
                "接通时长(秒)",
                "挂断者",
                "对话内容",
                "理解与回复能力",
                "原因",
                "情绪识别能力",
                "原因",
                "安抚能力",
                "原因",
                "ASR识别-转译错误",
                "原因",
            ]
            sheet_infos = [sheet_name, excel_title, excel_data]
            save_dir = "./excel"
            excel_file_name = f"{self.business}_大模型质检"
            excel_names = list()
            excel_name = write_excel2(sheet_infos, save_dir, excel_file_name)
            excel_names.append(excel_name)
            return {"code": 0, "excel_names": excel_names, "msg": None}


if __name__ == "__main__":
    ...
